package com.fyy.kafuka.interceptor;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.ArrayList;
import java.util.Properties;

/**
 * @Author fyy
 * @Description
 * @Date 2020/11/6 16:14
 */
public class InterceptorMain {

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"39.101.190.227:9092"); //指定连接集群
        properties.put(ProducerConfig.ACKS_CONFIG,"all"); //ack应答级别
        properties.put(ProducerConfig.RETRIES_CONFIG,"1"); //重试次数
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384); //发送批次大小16kb
        properties.put(ProducerConfig.LINGER_MS_CONFIG,10); //发送等待时间
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432); //缓冲区大小32M
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer"); //key-value的序列化类
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer"); //key-value的序列化类
        //使用自定义拦截器
        ArrayList<String> interceptors = new ArrayList<>();
        interceptors.add("com.fyy.kafuka.interceptor.AddTimeInterceptor");
        interceptors.add("com.fyy.kafuka.interceptor.CountInterceptor");
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,interceptors);
        //实例化一个生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        //发送数据(指定分区,指定key)
        producer.send(new ProducerRecord<>("test",0,"test", "ceshi"));
        //关闭连接
        producer.close();
    }

}
